package com.ibm.spatiotemp.index;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

import com.ibm.spatiotemp.datamodel.datatype.Constraints;
import com.ibm.spatiotemp.server.config.LocalConfig;

import edu.umn.cs.spatialHadoop.io.MemoryInputStream;

public class GeoIndexDBManager {
	
	public static Logger logger = Logger.getLogger(GeoIndexDBManager.class);
	
	public static FileSystem getFileSystemByPath(String path) throws IOException{
		FileSystem fs = null;
		Path inFile = new Path(path);
		fs = inFile.getFileSystem(new Configuration());
		inFile = null;
		return fs;
	}
	
	
	public static FSDataInputStream getRtreeDataInputStream(String rtreepath_path_str, boolean loadtocache) throws IOException
	{
		IndexDataLoader idxlocator = IndexDataLoaderFactory.getIndexDataLoaderInstance(rtreepath_path_str);
		FSDataInputStream fs_instream = null;
		FileSystem fs = null;
		
		int idxtag = idxlocator.locateIndexByPathKey(rtreepath_path_str,loadtocache);
		switch(idxtag)
		{
		case Constraints.INDEXLOCATION_HDFS:{
				Path inFile = new Path(rtreepath_path_str);
				fs = getFileSystemByPath(rtreepath_path_str);
				fs_instream = fs.open(inFile);
				/*xuxiaomin: SpatialIndex Header 8bytes skipped, see RTree.java*/
				(fs_instream).seek(8);
				break;
		}
		case Constraints.INDEXLOCATION_LOCALMEM:{
			/*means already existing in localmemory as byte[], to be implemnted later*/
			throw new RuntimeException("local memory index, not implemented");
		}
		case Constraints.INDEXLOCATION_REDIS:{
			logger.info("ready to read index data from redis...");
			if(LocalConfig.RunningLevel == LocalConfig.RunningLevel_Debug){
				rtreepath_path_str = LocalConfig.RtreePath4Debug;
				logger.info("only see in debug mode..., rtreepath_path_str="+rtreepath_path_str);
			}
			byte[] res = idxlocator.loadIndexByPathKey(rtreepath_path_str);
			if(res == null){
				logger.warn("read from redis failed, key ttl expired");
				Path inFile = new Path(rtreepath_path_str);
				fs = getFileSystemByPath(rtreepath_path_str);
				fs_instream = fs.open(inFile);
				/*xuxiaomin: SpatialIndex Header 8bytes skipped, see RTree.java*/
				(fs_instream).seek(8);
				logger.info("loaded from hdfs successfully");
				break;
			}
			else{
				fs_instream = new FSDataInputStream(new MemoryInputStream(res));
				/*xuxiaomin: SpatialIndex Header 8bytes skipped, see RTree.java*/
				(fs_instream).seek(8);
				break;
			}
		}
		}
		idxlocator.close();
		return fs_instream;

	}
}
